//*********************************************************
//
//    Copyright (c) Microsoft. All rights reserved.
//    This code is licensed under the Apache License, Version 2.0.
//    THIS CODE IS PROVIDED *AS IS* WITHOUT WARRANTY OF
//    ANY KIND, EITHER EXPRESS OR IMPLIED, INCLUDING ANY
//    IMPLIED WARRANTIES OF FITNESS FOR A PARTICULAR
//    PURPOSE, MERCHANTABILITY, OR NON-INFRINGEMENT.
//
//*********************************************************

namespace Microsoft.Research.eResearch.Blackboard
{
    using System;
    using System.Collections.Generic;
    using System.Configuration;
    using System.IO;
    using System.Linq;
    using System.Runtime.Serialization;
    using System.Runtime.Serialization.Formatters.Binary;
    using System.Threading;
    using Microsoft.Research.eResearch.Blackboard.Common;
    using Microsoft.Research.eResearch.Blackboard.Interfaces;
    using Microsoft.Research.eResearch.Common.Eventing;
    using Microsoft.Research.eResearch.Common.Threading;

    /// <summary>
    /// Provides an ISubscriptionStore that allows storage of all information about
    /// the current subscribers and their subscription information.  It is implemented
    /// as in-memory storage that is binary-serialized to disk whenever it changes.
    /// </summary>
    public class FileSubscriptionStore : ISubscriptionStore
    {
        /// <summary>
        /// Name of the sbscription store token.
        /// </summary>
        private const string SubscriptionStoreToken = "SubscriptionStore";

        /// <summary>
        /// Name of the file name token.
        /// </summary>
        private const string DefaultFileName = "SubscriptionStore.bin";

        /// <summary>
        /// Instance of the reader writer lock.
        /// </summary>
        private ReaderWriterLockSlim subscriberStoreLock = new ReaderWriterLockSlim();

        /// <summary>
        /// SubscriberStore - Consists of a dictionary of all current subscriber information:
        /// Key = Subscriber return address
        /// Value = Subscription profile (Key and CodeDom pair).
        /// </summary>
        private Dictionary<Uri, SubscriptionProfile> subscriberStore;

        /// <summary>
        /// AggregateSubscriptionProfile - The set theoretic union of key values in each of the
        /// subscription profiles.
        /// </summary>
        private SubscriptionProfile aggregateSubscriptionProfile;

        /// <summary>
        /// SubscriberLookupIndex - Consists of the list of keys and the list of subscribers that
        /// are interested in it.
        /// </summary>
        /// <remarks>
        /// This data structure enables faster lookup of subscribers registered for a specific key
        /// in their subscription profile.
        /// </remarks>
        private Dictionary<string, List<Uri>> subscriberLookupIndex;

        /// <summary>
        /// The filename where the SubscriptionStore will be serialized.
        /// </summary>
        private string fileName;

        /// <summary>
        /// Initializes a new instance of the <see cref="FileSubscriptionStore"/> class.
        /// The constructor also attempts to restore from a saved file if one exists.
        /// </summary>
        public FileSubscriptionStore()
        {
            // Get the filename from the config
            fileName = ConfigurationManager.AppSettings[SubscriptionStoreToken];

            if (String.IsNullOrEmpty(fileName))
            {
                fileName = DefaultFileName;
                ConfigurationManager.AppSettings[SubscriptionStoreToken] = DefaultFileName;
            }

            LoadSubscriberStoreFromDisk();
        }

        /// <summary>
        /// Raised when the subscription store changes (i.e. when a subscriber subscribers or unsubscribes).
        /// This event is only raised if a subscriber subscription/unsubscription changes in such a way that all
        /// publishers must be notified of a new aggregate subscription profile.
        /// </summary>
        public event EventHandler<EventArgs> SubscriptionStoreChanged;

        /// <summary>
        /// Gets the filename where the SubscriptionStore will be serialized.
        /// </summary>
        public string FileName
        {
            get { return fileName; }
        }

        /// <summary>
        /// This method is used to add a new subscriber to the blackboard
        /// or to update the subscription profile of an existing subscriber.
        /// </summary>
        /// <remarks>
        /// If the subscriber is new, then three data structures are updated:
        /// 1. SubscriberStore and written to file (disk)
        /// 2. SubscriberLookupIndex
        /// 3. AggregateSubscriptionProfile
        /// If existing subscriber sends an updated profile, a 'diff' between the current profile and new profile is used to
        /// update the SubscriberLookupIndex, which appropriately affects the flags.
        /// </remarks>
        /// <param name="returnAddress">The URL for the subscriber to receive callbacks.</param>
        /// <param name="profile">The subscriber's requested subscription profile.</param>
        public void Subscribe(Uri returnAddress, SubscriptionProfile profile)
        {
            bool subscriberStoreChanged = false;
            bool subscriberLookupIndexChanged = false;

            using (new AcquireWriterLock(subscriberStoreLock))
            {
                // String to store list of key values to be removed from SubscriberLookupIndex
                List<string> sliKeyRemoveList = new List<string>();

                // 1. SubscriberStore - Insert a new value or replace existing entry for give subscriber in the SubscriberStore
                subscriberStore[returnAddress] = profile;

                // 2. Check if the SubscriberLookupIndex needs to changed - appropriate flags are set to fire events for:
                // a. Updating the AggregateSubscriptionProfile and send messages to each publisher
                // b. Write back the SubscriberStore to the disk
                foreach (string fieldName in subscriberLookupIndex.Keys)
                {
                    // Retrieve the key (SubscriptionItem) from the profile for given fieldName
                    SubscriptionItem keyValue = profile[fieldName];

                    // I. Check if key name is present in new profile of an existing subscriber
                    // If the SubscriberLookupIndex contains key also present in new subscriber profile remove it
                    if (keyValue != null)
                    {
                        // Step 1
                        // Check if returnAddress of the subscriber is present for the given key
                        if (!subscriberLookupIndex[fieldName].Contains(returnAddress))
                        {
                            // Add the subscriber URI to the list
                            subscriberLookupIndex[fieldName].Add(returnAddress);

                            // Set the flag that subscriber store needs to be persisted                            
                            subscriberLookupIndexChanged = true;
                        } // End of inner if

                        // Step 2                        
                        // Remove the key from the new Subscriber profile - the remaining keys in the profile will be added SubscriberLookupIndex                        
                    }
                    else
                    {
                        // Check if the URI of the subscriber is in the list for the given key value
                        List<Uri> list = subscriberLookupIndex[fieldName];
                        if (list.Contains(returnAddress))
                        {
                            // Delete the URI from the list for given key value
                            list.Remove(returnAddress);

                            // Check if there are any subscribers left in the list for given key, if list length is 0, delete the entry from SubscriberLookupIndex
                            if (list.Count == 0)
                            {
                                // Add the key value to the list of keys to be removed from SubscriberLookupIndex
                                sliKeyRemoveList.Add(fieldName);

                                // Set the flag for change in SubscriberLookupIndex
                                subscriberLookupIndexChanged = true;
                            } // End of inner if                        
                        }
                    } // End of else-if                    
                }

                // II. Remove all the keys with 0 subscribers from the SubscriberLookupIndex
                sliKeyRemoveList.ForEach(fieldName => subscriberLookupIndex.Remove(fieldName));

                // III. Add the remaining key values in the new subscriber profile to the SubscriberLookupIndex
                foreach (SubscriptionItem item in profile)
                {
                    // if the lookup index does not contain the concept then add it.
                    // concept existing in lookup index scenario is already handled by the above code
                    if (!subscriberLookupIndex.ContainsKey(item.Name))
                    {
                        // Add the key and the URI of the subscriber to the SubscriberLookupIndex
                        subscriberLookupIndex.Add(item.Name, new List<Uri>() { returnAddress });

                        // Set the subscriberStore changed flag
                        subscriberStoreChanged = true;
                    }
                }

                if (subscriberStoreChanged || subscriberLookupIndexChanged)
                {
                    WriteSubscriberStoreToDisk();
                }

                if (subscriberStoreChanged)
                {
                    RebuildAggregateSubscriptionProfile();
                }
            }

            if (subscriberStoreChanged)
            {
                OnSubscriptionStoreChanged();
            }
        }

        /// <summary>
        /// Deletes an existing subscriber from the blackboard.
        /// </summary>
        /// <remarks>
        /// Requires updation of three data structures:
        /// 1. SubscriberStore
        /// 2. AggregateSubscriptionProfile
        /// 3. SubscriberLookupIndex
        /// </remarks>
        /// <param name="returnAddress">The subscriber's return address to unsubscribe.</param>
        public void Unsubscribe(Uri returnAddress)
        {
            bool subscriberStoreChanged = false;
            bool subscriberRemoved = false;

            using (new AcquireWriterLock(subscriberStoreLock))
            {
                // Check if subscriber exists on the SubscribersStore, if yes get its subscription profile
                if (subscriberStore.ContainsKey(returnAddress))
                {
                    // Get the profile of the given subscriber if it exists
                    SubscriptionProfile sp = subscriberStore[returnAddress];

                    // Delete the entry for the given subscription profile in SubscriberStore
                    subscriberStore.Remove(returnAddress);

                    // Set the flag - to write subscriber store to disk
                    subscriberStoreChanged = true;

                    // 2. Update the SubscriberLookupIndex to reflect the deletion of a subscriber - reduce the count of subscribers for a given key or delete the key from the SubscriberLookupIndex
                    foreach (SubscriptionItem item in sp)
                    {
                        // SubscriberLookupIndex is updated according the keys specified in the subscription profile of the deleted subscriber
                        // 1. If the key is present, delete the subscriber from the list of subscriber for that key
                        // 2. If the final count of the subscribers for the given key is 0, delete the key from the SubscriberLookupIndex
                        // This updated state of the SubscriberLookupIndex is used to update the AggregateSubscriptionProfile
                        if (subscriberLookupIndex.ContainsKey(item.Name))
                        {
                            // Create a new list for optimization
                            List<Uri> list = subscriberLookupIndex[item.Name];

                            // Check if the subscriber URI is present in the list
                            if (list.Contains(returnAddress))
                            {
                                // Remove the uri from list of subscribers for given key
                                list.Remove(returnAddress);

                                // Check if there are any subscribers left in the list for given key, if list length is 0, delete the entry from SubscriberLookupIndex
                                if (list.Count == 0)
                                {
                                    // Remover the entry for given key from the SubscriberLookupIndex
                                    subscriberLookupIndex.Remove(item.Name);

                                    subscriberRemoved = true;
                                }
                            }
                        }
                    }
                }

                if (subscriberStoreChanged || subscriberRemoved)
                {
                    WriteSubscriberStoreToDisk();
                }

                if (subscriberStoreChanged)
                {
                    RebuildAggregateSubscriptionProfile();
                }
            }

            if (subscriberStoreChanged)
            {
                OnSubscriptionStoreChanged();
            }
        }

        /// <summary>
        /// Gets the aggregate subscription profile that the publisher can use.
        /// This profile is a set union of all of the items that subscribers have requested.
        /// </summary>
        /// <returns>The aggregate subscription profile.</returns>
        public SubscriptionProfile GetAggregateProfile()
        {
            SubscriptionProfile copy;
            using (new AcquireReaderLock(subscriberStoreLock))
            {
                copy = new SubscriptionProfile(aggregateSubscriptionProfile);
            }

            return copy;
        }

        /// <summary>
        /// This method takes an incoming BlackboardMessage and breaks up the name-value-pairs
        /// into distinct BlackboardMessages for each subscriber, depending on the subscriber's
        /// SubscriptionProfile.  
        /// </summary>
        /// <param name="incomingMessage">The incoming message.</param>
        /// <returns>A collection of BlackboardMessages and their destination URLs.</returns>
        public IDictionary<Uri, BlackboardMessage> CreateNotificationMessages(BlackboardMessage incomingMessage)
        {
            Dictionary<Uri, BlackboardMessage> outgoingMessages = new Dictionary<Uri, BlackboardMessage>();

            using (new AcquireReaderLock(subscriberStoreLock))
            {
                foreach (KeyValuePair<string, string> item in incomingMessage)
                {
                    if (subscriberLookupIndex.ContainsKey(item.Key))
                    {
                        List<Uri> subscribers = subscriberLookupIndex[item.Key];
                        foreach (Uri subscriber in subscribers)
                        {
                            if (outgoingMessages.ContainsKey(subscriber))
                            {
                                outgoingMessages[subscriber].Add(item.Key, item.Value);
                            }
                            else
                            {
                                BlackboardMessage message = new BlackboardMessage();
                                message.Add(item.Key, item.Value);
                                outgoingMessages.Add(subscriber, message);
                            }
                        }
                    }
                }
            }

            return outgoingMessages;
        }

        /// <summary>
        /// Raises the SubscriptionStoreChanged event.
        /// </summary>
        protected virtual void OnSubscriptionStoreChanged()
        {
            this.SubscriptionStoreChanged.Fire(this, EventArgs.Empty);
        }

        /// <summary>
        /// Writes the subscriber store to disk.
        /// </summary>
        private void WriteSubscriberStoreToDisk()
        {
            try
            {
                using (Stream fs = new FileStream(fileName, FileMode.Create))
                {
                    IFormatter formatter = new BinaryFormatter();
                    formatter.Serialize(fs, subscriberStore);
                }
            }
            catch (Exception)
            {
                // Nothing we can really do, so just eat the exception.
                // Everything will still work since the data is also in memory.
            }
        }

        /// <summary>
        /// Loads the subscriber store from disk.
        /// </summary>
        private void LoadSubscriberStoreFromDisk()
        {
            try
            {
                using (Stream fs = new FileStream(fileName, FileMode.Open))
                {
                    IFormatter formatter = new BinaryFormatter();
                    subscriberStore = (Dictionary<Uri, SubscriptionProfile>)formatter.Deserialize(fs);
                    CreateSubscriberLookupIndex();
                    RebuildAggregateSubscriptionProfile();
                }
            }
            catch (FileNotFoundException)
            {
                subscriberStore = new Dictionary<Uri, SubscriptionProfile>();
                subscriberLookupIndex = new Dictionary<string, List<Uri>>();
                aggregateSubscriptionProfile = new SubscriptionProfile();
                WriteSubscriberStoreToDisk();
            }
        }

        /// <summary>
        /// Creates the initial Subscriber Lookup Index after a new SubscriptionStore has been loaded.
        /// </summary>
        private void CreateSubscriberLookupIndex()
        {
            subscriberLookupIndex = new Dictionary<string, List<Uri>>();
            foreach (KeyValuePair<Uri, SubscriptionProfile> subscriber in subscriberStore)
            {
                foreach (SubscriptionItem item in subscriber.Value)
                {
                    if (subscriberLookupIndex.ContainsKey(item.Name))
                    {
                        List<Uri> list = subscriberLookupIndex[item.Name];
                        if (!list.Contains(subscriber.Key))
                        {
                            list.Add(subscriber.Key);
                        }
                    }
                    else
                    {
                        subscriberLookupIndex.Add(item.Name, new List<Uri>() { subscriber.Key });
                    }
                }
            }
        }

        /// <summary>
        /// Rebuilds the aggregate subscription profile.
        /// </summary>
        private void RebuildAggregateSubscriptionProfile()
        {
            // Use the set of keys in the current SubscriberLookupIndex to create an aggregate subscription profile
            aggregateSubscriptionProfile =
                new SubscriptionProfile(subscriberLookupIndex.Keys.Select(item => new SubscriptionItem(item)));
        }
    }
}
